package com.learning.thread.juc.callable;

import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.AsyncFunction;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListenableFutureTask;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * ClassName: CallableGuava
 * Description: Guava 的 Futures 框架
 * Date: 2016/11/26 17:45
 *
 * @author SAM SHO
 * @version V1.0
 */
public class CallableGuava {


    /**
     * 提交一个Callable
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    void test() {
        ExecutorService executor = Executors.newFixedThreadPool(1);
        // 提交一个 Callable
        Future<String> future = executor.submit(new InCallableTask());

        // 会一直阻塞, 直到任务完成，结果返回
        try {
            future.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }


    /**
     * 使用 ListenableFuture
     * Future 通过异步的方式计算返回结果
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public void test1() throws ExecutionException, InterruptedException {

        // 自定义线程池名称，使用 ThreadFactory 的构造方法
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("guava-ListenableFuture-%s").build();
        ExecutorService executorService = Executors.newFixedThreadPool(10, threadFactory);

        // 将 ExecutorService 转为 ListeningExecutorService
        ListeningExecutorService service = MoreExecutors.listeningDecorator(executorService);
        // 提交任务
        ListenableFuture<String> listenableFuture = service.submit(new InCallableTask());

        // 可以注册一个回调, 在回调函数中对异步处理的结果进行处理
        // 这个方法与 JDK 中的 future.get() 不同，不会 callback() 函数不阻塞当前线程
        Futures.addCallback(listenableFuture, new FutureCallback<String>() {
            // we want this handler to run immediately after we push the big red button!
            @Override
            public void onSuccess(String result) {
                System.out.println("获取的结果 " + result);
            }

            @Override
            public void onFailure(Throwable thrown) {
                thrown.printStackTrace();
            }
        });

        // 上面任务的异常，不会影响主线程后续的执行
        System.out.println("这边会先执行？ main task done.....");

        MoreExecutors.shutdownAndAwaitTermination(service, 60, TimeUnit.SECONDS);
    }

    /**
     * 针对JDK 的 FutureTask 转换而来的, Guava 提供ListenableFutureTask.create(Callable<V>) 和ListenableFutureTask.create(Runnable, V)
     */
    public void test2() {
        // 自定义线程池名称，使用 ThreadFactory 的构造方法
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("guava-ListenableFutureTask-%s").build();
        ThreadPoolExecutor executorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), threadFactory);

        // 将 ThreadPoolExecutor 转为 ListeningExecutorService
        ListeningExecutorService service = MoreExecutors.listeningDecorator(executorService);

        // 创建 ListenableFutureTask
        ListenableFutureTask<String> listenableFutureTask = ListenableFutureTask.create(new InCallableTask());
        // 提交任务
        service.submit(listenableFutureTask);

        Futures.addCallback(listenableFutureTask, new FutureCallback<String>() {
            @Override
            public void onSuccess(String result) {
                System.out.println("获取的结果： " + result);
            }

            @Override
            public void onFailure(Throwable thrown) {
                thrown.printStackTrace();
            }
        });

        System.out.println("这边会先执行？ main task done.....");

        MoreExecutors.shutdownAndAwaitTermination(service, 60, TimeUnit.SECONDS);
    }

    /**
     * 提交一组任务
     */
    public void test3() throws ExecutionException, InterruptedException {
        // 自定义线程池名称，使用 ThreadFactory 的构造方法
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("guava-ListenableFutureTask-%s").build();
        ThreadPoolExecutor executorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), threadFactory);

        // 将 ThreadPoolExecutor 转为 ListeningExecutorService
        ListeningExecutorService service = MoreExecutors.listeningDecorator(executorService);


        // 任务列表
        ArrayList<Callable<String>> tasks = Lists.newArrayList(new InCallableTask(), new InCallableTaskLongTime());

        List<ListenableFuture<String>> futures = Lists.newArrayList();

        // 提交任务
        for (Callable<String> task : tasks) {
            futures.add(service.submit(task));
        }
        // 返回一个ListenableFuture ，该Future的结果包含所有成功的Future，按照原来的顺序，当其中之一Failed或者cancel，则用null替代
        ListenableFuture<List<String>> listListenableFuture = Futures.successfulAsList(futures);

        // 堵塞，直到所有任务完成，然后一起返回
        // [Hello SAM!!  guava-ListenableFutureTask-0, Hello SAM, long time task guava-ListenableFutureTask-0]
        // 如果一个任务发生异常: [null, Hello SAM, long time task guava-ListenableFutureTask-0]
        System.out.println(listListenableFuture.get());

        // allAsList与 successfulAsList的区别
        // 如果一个任务发生异常, 这边会抛出异常，
        ListenableFuture<List<String>> listListenableFuture1 = Futures.allAsList(futures);
        System.out.println(listListenableFuture1.get());

        System.out.println("这边不会先执行 main task done.....");

        // 一定要记得关闭
        MoreExecutors.shutdownAndAwaitTermination(service, 60, TimeUnit.SECONDS);
    }

    /**
     * Guava 的 catching
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public void test4() throws ExecutionException, InterruptedException {
        // 自定义线程池名称，使用 ThreadFactory 的构造方法
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("guava-ListenableFutureTask-%s").build();
        ThreadPoolExecutor executorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), threadFactory);

        // 将 ThreadPoolExecutor 转为 ListeningExecutorService
        ListeningExecutorService service = MoreExecutors.listeningDecorator(executorService);

        final InCallableTask inCallableTask = new InCallableTask();
        ListenableFuture<String> listenableFuture = service.submit(inCallableTask);

        Futures.catching(listenableFuture, RuntimeException.class, new Function<RuntimeException, String>() {
            @Override
            public String apply(RuntimeException input) {
                // 这边可以做一些其他事情,，如重新调用，各种异常处理
                return "你好，异常被我吃了";
            }
        });

        Futures.catchingAsync(listenableFuture, RuntimeException.class, new AsyncFunction<RuntimeException, String>() {
            @Override
            public ListenableFuture<String> apply(RuntimeException input) throws Exception {
                return listenableFuture;
            }
        });

        System.out.println(listenableFuture.get());
    }

    class InCallableTask implements Callable<String> {

        @Override
        public String call() throws Exception {
            Thread.sleep(2000L);
            // 异常测试
            int i = 1 / 0;
            return "Hello SAM!!  " + Thread.currentThread().getName();
        }
    }


    class InCallableTaskLongTime implements Callable<String> {

        @Override
        public String call() throws Exception {
            Thread.sleep(10000L);
            return "Hello SAM, long time task " + Thread.currentThread().getName();
        }
    }
}
